/**
 * Software License, Version 1.0
 * 
 * Copyright 2008 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgment:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgment may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or Granules, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called Granules, 
 * nor may Indiana University or Community Grids Lab or Granules appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 * LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 * WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 * NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 * INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 * INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 * "VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 * LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 * ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 * GENERATED USING SOFTWARE.
 */
package cgl.granules.scheduler;

import java.io.BufferedOutputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Properties;

import cgl.granules.dataset.DatasetCollection;
import cgl.granules.dataset.DatasetCollectionObserver;
import cgl.granules.dataset.DatasetException;
import cgl.granules.operation.ExecutionProfile;
import cgl.granules.operation.Operation;
import cgl.granules.operation.ProcessingException;

/**
 * This is an executable task, which manages the execution of the operation. It
 * allows the registration of a taskLifeTimeObserver so that the observer can
 * track the progress of the execution, and act upon it.
 * 
 * @author Shrideep Pallickara
 * 
 */
public class ExecutableTask implements Runnable, DatasetCollectionObserver {
	private String taskIdentifier;
	private Operation operation;
	private TaskLifecycleObserver taskLifetimeObserver;
	private TaskLifecycleMetricsImpl taskLifecycleMetrics;
	private ExecutionProfile executionProfile;
	private boolean executing = false;

	private int numOfIterations = 0;
	private long executionStartTime = 0;

	private DatasetCollection datasetCollection;

	private boolean periodic;
	private long periodicInterval;
	private boolean debug = false;
	private long initialDelayForPeriodicTask = 0;

	protected ExecutableTask(String taskIdentifier, Operation operation,
			TaskLifecycleObserver taskLifetimeObserver)
			throws ProcessingException, DatasetException {

		this.taskIdentifier = taskIdentifier;
		this.operation = operation;
		this.taskLifetimeObserver = taskLifetimeObserver;

		executionProfile = operation.getExecutionProfile();
		periodic = executionProfile.isExecutePeriodically();
		periodicInterval = executionProfile.getExecutionInterval();

		if (operation.needsInitialization()) {
			Properties processingDirectives = operation
					.getProcessingDirectives();
			operation.initialize(processingDirectives);
		}

		if (operation.hasDatasetCollection()) {
			datasetCollection = operation.getDatasetCollection();
			datasetCollection.setDatasetCollectionObserver(this);
		}
		taskLifecycleMetrics = new TaskLifecycleMetricsImpl(System
				.currentTimeMillis());
	}

	/**
	 * Retrieve the identifier for the executable task
	 * 
	 * @return The identifier for the executable task
	 */
	public String getIdentifier() {
		return taskIdentifier;
	}

	/**
	 * Retrieve the TaskLifecyleMetrics associated with this task.
	 * 
	 * @return The lifecyle metrics for this task
	 */
	public TaskLifecycleMetrics getTaskLifecycleMetrics() {
		return taskLifecycleMetrics;
	}

	/**
	 * Set the execution status for the lifecycle metrics.
	 * 
	 * @param executionStatus
	 */
	public void setExecutionStatus(int executionStatus) {
		taskLifecycleMetrics.setExecutionStatus(executionStatus);
	}

	/**
	 * Retrieve the operation that will be executed by this task
	 * 
	 * @return The operation being executed
	 */
	public Operation getOperation() {
		return operation;
	}

	@Override
	public void run() {
		try {
			executing = true;
			long startTime = System.nanoTime();
			if (executionStartTime == 0) {
				taskLifecycleMetrics.setStartTime();
			}
			taskLifecycleMetrics
					.setExecutionStatus(TaskLifecycleMetrics.EXECUTING);
			operation.execute();
			long cpuBoundTime = System.nanoTime() - startTime;
			numOfIterations++;
			taskLifecycleMetrics.updateMetrics(cpuBoundTime, numOfIterations);
			executing = false;
		} catch (Exception e) {
			String errorReport = e.getMessage();
			taskLifecycleMetrics
					.setExecutionStatus(TaskLifecycleMetrics.ABNORMAL_TERMINATION);
			taskLifetimeObserver.abnormalOperationTermination(taskIdentifier,
					errorReport);
			return;
		}

		if (isTaskComplete()) {
			taskLifecycleMetrics
					.setExecutionStatus(TaskLifecycleMetrics.COMPLETE);
			taskLifecycleMetrics.setCompletion();
			taskLifetimeObserver.isComplete(taskIdentifier);
			return;
		}

		if (isTaskPeriodicityChanged()) {
			periodic = executionProfile.isExecutePeriodically();
			periodicInterval = executionProfile.getExecutionInterval();
			initialDelayForPeriodicTask = periodicInterval;
			taskLifetimeObserver.isChangedPeriodicExecution(taskIdentifier);
		}

		if (taskLifecycleMetrics.getExecutionStatus() != TaskLifecycleMetrics.WAITING_FOR_DATA) {
			taskLifecycleMetrics
					.setExecutionStatus(TaskLifecycleMetrics.QUEUED_FOR_EXECUTION);
		}
		
		taskLifetimeObserver.isExecuting(taskIdentifier);//done executing
	}

	/**
	 * Indicates whether the operation passes the initialization check, and if
	 * the encapsulating task is ready for execution.
	 * 
	 * @return true: if the task is ready for execution for the <i>first</i>
	 *         time.
	 */
	private boolean passesInitializationCheck() {
		boolean initialized = true;
		if (operation.needsInitialization()) {
			if (!operation.isInitialized()) {
				String errorReport = "Operation " + operation
						+ " needs to be initialized. "
						+ "But the isInitialized() method returns false";
				System.out.println(errorReport);
				initialized = false;
			}
		}
		return initialized;
	}

	/**
	 * Checks to see if the task is ready for execution. A check is first made
	 * to see if the operation is initialized properly. Next a check is made to
	 * see if the execution is predicated on data availability, and if it is,
	 * make sure that data is available before deciding to proceed
	 * 
	 * @return true If the task is ready for execution
	 */
	public boolean passesReadinessCheck() {
		boolean readinessFlag = false;

		/*
		 * First check to see if the operation passes the initialization check
		 */
		boolean passInitializationCheck = passesInitializationCheck();
		if (!passInitializationCheck) {
			return readinessFlag;
		}

		/*
		 * Check if the execution is predicated on data availability, and if it
		 * is, make sure that data is available before deciding to proceed
		 */
		if (executionProfile.isExecuteWhenDataAvailable()) {
			if (datasetCollection.isDataAvailable()) {
				readinessFlag = true;
				if (debug) {
					System.out.println("Data is available!");
				}
			} else {
				readinessFlag = false;
				taskLifecycleMetrics
						.setExecutionStatus(TaskLifecycleMetrics.WAITING_FOR_DATA);
			}
			return readinessFlag;
		}
		readinessFlag = true;

		return readinessFlag;
	}

	/**
	 * Checks to see if the task is complete.
	 * 
	 * @return
	 */
	private boolean isTaskComplete() {
		boolean taskComplete = false;
		if (executionProfile.isExecuteOnce()) {
			taskComplete = true;
		}

		if (executionProfile.isStayAlive()
				|| executionProfile.isExecutePeriodically()) {
			if (operation.terminationConditionReached()) {
				taskComplete = true;
			}
		}

		if (executionProfile.isExcuteAFixedNumberOfTimes()) {
			int maxLimit = executionProfile.getNumberOfTimesToExecute();
			if (numOfIterations >= maxLimit) {
				taskComplete = true;
			}
		}

		return taskComplete;
	}

	/**
	 * This method returns true only if the Task that was previously periodic,
	 * is now changed to regular execution.
	 * 
	 * @return true if it has changed; false otherwise
	 */
	private boolean isTaskPeriodicityChanged() {
		if (periodic) {
			if (!executionProfile.isExecutePeriodically()) {
				return true;
			}

			if (periodicInterval != executionProfile.getExecutionInterval()) {
				if (debug) {
					System.out.println("PeriodicInterval =" + periodicInterval
							+ ", executionInterval="
							+ executionProfile.getExecutionInterval());
				}
				return true;
			}
		}

		return false;
	}

	@Override
	public void onDataAvailability() {

		if (executionProfile.isExecuteWhenDataAvailable()) {
			if (debug) {
				System.out.println("data available!");
			}

			taskLifetimeObserver.isReady(taskIdentifier);
		}

	}

	@Override
	public void onDatasetCollectionClosure() {
		if (executionProfile.isExecuteWhenDataAvailable()) {
			if (!executing) {
				taskLifetimeObserver.isComplete(taskIdentifier);
			}
		}
	}

	/**
	 * A check to see if the task in question needs to be executed periodically.
	 * 
	 * @return <code>true</code> if the task needs to be executed periodically,
	 *         and <code>false</code> otherwise.
	 */
	public boolean isExecutePeriodically() {
		return periodic;
	}

	/**
	 * For a task that needs to be executed periodically, this method is used to
	 * retrieve the interval at which the task needs to be executed
	 * periodically.
	 * 
	 * @return The interval (in milliseconds) for periodic execution.
	 */
	public long getExecutionInterval() {
		return periodicInterval;
	}

	/**
	 * This method returns the time at which the periodic task will execute for
	 * the first time.
	 * 
	 * @return The delay for executing periodic task for the first time.
	 */
	public long getInitialDelayForPeriodicTask() {
		return initialDelayForPeriodicTask;
	}

	@Override
	public int hashCode() {
		return taskIdentifier.hashCode();
	}

	@Override
	public boolean equals(Object anObject) {
		if (this == anObject) {
			return true;
		}
		if (anObject instanceof ExecutableTask) {
			ExecutableTask anotherTask = (ExecutableTask) anObject;
			if (taskIdentifier.equals(anotherTask.getIdentifier())) {
				return true;
			}
		}
		return false;
	}

	class TaskLifecycleMetricsImpl implements TaskLifecycleMetrics {
		private volatile long arrivalTime, cpuBoundTime, completionTime,
				numberOfIterationsSoFar, startTime, timeResidentInMemory;
		private int executionStatus = QUEUED_FOR_EXECUTION;

		private TaskLifecycleMetricsImpl(long arrivalTime) {
			this.arrivalTime = arrivalTime;
		}

		public int getExecutionStatus() {
			return executionStatus;
		}

		private void setExecutionStatus(int executionStatus) {
			this.executionStatus = executionStatus;
		}

		private void setStartTime() {
			startTime = System.currentTimeMillis();
		}

		private void updateMetrics(long cpuBound, int numOfIterations) {
			cpuBoundTime += cpuBound;
			numberOfIterationsSoFar = numOfIterations;
		}

		private void setCompletion() {
			completionTime = System.currentTimeMillis();
			executionStatus = COMPLETE;
		}

		@Override
		public long getArrivalTime() {
			return arrivalTime;
		}

		@Override
		public long getCPUBoundTime() {
			return cpuBoundTime;
		}

		@Override
		public long getCompletionTime() {
			return completionTime;
		}

		@Override
		public int getNumberofIterationsSoFar() {
			return (int) numberOfIterationsSoFar;
		}

		@Override
		public long getStartTime() {
			return startTime;
		}

		@Override
		public long getTimeResidentInMemory() {
			return timeResidentInMemory;
		}

		@Override
		public String toString() {
			String stringRepresentation = "Execution Status=";
			switch (executionStatus) {
			case COMPLETE:
				stringRepresentation += "(COMPLETE) ";
				break;
			case EXECUTING:
				stringRepresentation += "(Executing) ";
				break;
			case WAITING_FOR_DATA:
				stringRepresentation += "(Waiting For Data) ";
				break;
			case QUEUED_FOR_EXECUTION:
				stringRepresentation += "(Queued For Execution) ";
				break;
			default:
				stringRepresentation += ("(Unknown = " + executionStatus + " ");
				break;
			}

			stringRepresentation += " Arrival Time=" + arrivalTime
					+ ", Start Time =" + startTime + ", CPU Bound Time = "
					+ cpuBoundTime + " nanoSecs";

			if (numOfIterations > 1) {
				stringRepresentation += (", Number Of Iterations =" + numOfIterations);
			}

			if (executionStatus == COMPLETE) {
				stringRepresentation += (", Completion Time =" + completionTime);
			}
			return stringRepresentation;
		}

		@Override
		public byte[] getBytes() throws IOException {
			byte[] marshalledBytes = null;
			ByteArrayOutputStream baOutputStream = new ByteArrayOutputStream();
			DataOutputStream dout = new DataOutputStream(
					new BufferedOutputStream(baOutputStream));

			dout.writeLong(arrivalTime);
			dout.writeLong(cpuBoundTime);
			dout.writeLong(completionTime);
			dout.writeLong(startTime);
			dout.writeLong(timeResidentInMemory);

			dout.writeInt((int) numberOfIterationsSoFar);
			dout.writeInt(executionStatus);

			dout.flush();
			marshalledBytes = baOutputStream.toByteArray();
			return marshalledBytes;
		}

	}

}
